/*
 * TOP SECRET Copyright 2006-2015 Transsion.com All right reserved. This software is the confidential and proprietary
 * information of Transsion.com ("Confidential Information"). You shall not disclose such Confidential Information and
 * shall use it only in accordance with the terms of the license agreement you entered into with Transsion.com.
 */
package com.palmplaystore.bigdata.storm.appupdate;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.palmplaystore.bigdata.ContextHolder;
import com.palmplaystore.bigdata.storm.appupdate.bean.AppObject;
import com.palmplaystore.bigdata.storm.appupdate.bean.AppsObject;
import com.palmplaystore.bigdata.storm.appupdate.bean.PackageObject;
import com.palmplaystore.bigdata.storm.appupdate.cache.CacheService;
import com.palmplaystore.bigdata.util.MD5Util;
import com.palmplaystore.bigdata.util.StringUtil;

/**
 * ClassName:AppSplitBolt <br/>
 * Date: 2018年9月3日 下午4:20:10 <br/>
 * 
 * @author fenglibin
 * @version
 * @see
 */
public class AppSplitBolt extends BaseRichBolt {

    private static final long serialVersionUID = -4561021970203940510L;
    private Logger            logger           = LoggerFactory.getLogger(AppSplitBolt.class);

    OutputCollector           collector;
    AppHBaseService appHBaseService;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        appHBaseService = ContextHolder.getApplicationContext().getBean(AppHBaseService.class);
    }

    @Override
    public void execute(Tuple input) {
        String appUpdateInfo = input.getString(4);
//        logger.info("Got appUpdateInfo:" + appUpdateInfo);
        List<AppObject> appList = handleRow(appUpdateInfo);
        if (appList != null&&appList.size()>0) {
        	collector.emit(input,new Values(appList));
        }
        collector.ack(input);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("app"));
    }

    /**
     * 将app更新列表拆开为多个实际的app更新列表
     * 
     * @param jsonString
     * @return
     */
    private List<AppObject> handleRow(String jsonString) {
        if (jsonString == null || jsonString.trim().length() == 0) {
            return null;
        }
        List<AppObject> list = new ArrayList<AppObject>();
        AppsObject appsObject = null;
        try {
            appsObject = JSONObject.parseObject(jsonString, AppsObject.class);
        } catch (Exception e) {
            saveErrJsonMessage(jsonString);
            return null;
        }
        if(StringUtil.isEmpty(appsObject.getImei())||!StringUtil.isInteger(appsObject.getImei())) {
        	return null;
        }
        AppObject origAppObject = new AppObject();
        origAppObject.setBrand(appsObject.getBrand());
        origAppObject.setChannel(appsObject.getChannel());
        origAppObject.setCurClientVersionCode(appsObject.getCurclientversioncode());
        origAppObject.setImei(appsObject.getImei());
        origAppObject.setImsi(appsObject.getImsi());
        origAppObject.setLan(appsObject.getLan());
        origAppObject.setNetType(appsObject.getNettype());
        origAppObject.setSystemVersionCode(appsObject.getCurclientversioncode());
        origAppObject.setUa(appsObject.getUa());

        origAppObject.setCountryCode(appsObject.getCountrycode());
        origAppObject.setOperator(appsObject.getOperator());
        origAppObject.setIpAddress(appsObject.getIpaddress());
        origAppObject.setCreateTime(appsObject.getCreatetime());
        origAppObject.setIsUpdate(appsObject.getIsUpdate());
        List<PackageObject> appsList = null;
        String apps = appsObject.getApps();
        try {
            appsList = JSONObject.parseArray(apps, PackageObject.class);
            List<PackageObject> newAppsList = new ArrayList<PackageObject>();
            for (PackageObject packageObject : appsList) {
                // 过滤掉那些不需要处理的app
                if (!CacheService.isExcludePackage(packageObject.getPackageName())) {
                    newAppsList.add(packageObject);
                }
            }
            appsList.clear();
            appsList = null;
            appsList = newAppsList;
        } catch (Exception e) {
            // 解析原始的app更新列表出错，因为上报上来的app列表数据上有问题
            List<String> appList = StringUtil.getListFromStart2End(apps, "{", "}", true);
            for (String appStr : appList) {
                try {
                    PackageObject packageObject = JSONObject.parseObject(appStr, PackageObject.class);
                    // 过滤掉那些不需要处理的app
                    if (!CacheService.isExcludePackage(packageObject.getPackageName())) {
                        appsList.add(packageObject);
                    }
                } catch (Exception e1) {
                    // 这里的Json串有些还是有问题的，针对这种情况，直接抛弃
                }
            }
        }
        int keyIndex = 1;
        if (appsList != null && appsList.size() > 0) {
            for (PackageObject item : appsList) {
                AppObject appObject = new AppObject();
                appObject.setImei(origAppObject.getImei());
                appObject.setKeyIndex(keyIndex);
                keyIndex++;
                appObject.setPackageName(item.getPackageName());
                appObject.setVersion(item.getVersion());
                list.add(appObject);
            }
            /**
             * 将所有可用的app全部组成一个新的对象，处理的时候根据version为-1进行判断处理
             */
            AppObject allAppsObject = origAppObject.clone();
            allAppsObject.setPackageName(JSONArray.toJSONString(appsList));
            allAppsObject.setVersion(-1);
            String mdtstr=allAppsObject.getImei()+allAppsObject.getCurClientVersionCode()+allAppsObject.getImsi()+allAppsObject.getUa()+allAppsObject.getChannel()+allAppsObject.getNetType()+allAppsObject.getBrand()+allAppsObject.getSystemVersionCode()+allAppsObject.getLan()+allAppsObject.getCountryCode()+allAppsObject.getOperator()+allAppsObject.getPackageName()+allAppsObject.getIsUpdate();
			String md5=MD5Util.md5Hex(mdtstr);
            allAppsObject.setMd5(md5);
            list.add(allAppsObject);
        }

        return list;
    }
    
    /**
     * 将不能够处理的JSON数据放到另外的Kafka队列中
     * @param message
     */
    private void saveErrJsonMessage(String message) {
        //todo
    }

}
